package net.xdclass.xdclasskafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

/**
 * @PACKAGE_NAME: net.xdclass.xdclasskafka.controller
 * @PROJECT_NAME: jz_dm_gateway
 * @NAME: UserController
 * @USER: zhangc
 * @DATE: 2021/11/14
 * @TIME: 22:27
 * @DESCRIPTION:
 **/
@RestController
public class UserController {
    private static final String TOPIC_NAME = "user.register.topic";
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 发送消息
     *
     * @param phone
     */
    @GetMapping("/api/user/{phone}")
    public void sendMessage(@PathVariable("phone") String phone) {
        kafkaTemplate.send(TOPIC_NAME, phone).addCallback(
                success -> {
                    //消息发送到topic
                    String topic = success.getRecordMetadata().topic();
                    //消息发送的分区
                    int partition = success.getRecordMetadata().partition();
                    //消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();
                    System.out.println("消息发送成功:"+topic+"_"+partition + "_"+offset);
                }, failure -> {
                    System.out.println("消息发送失败:"+failure.getMessage());
                });
    }
    /**
     * 注解方式的事务
     * @param num
     */
    @GetMapping("/api/v1/tran1")
    @Transactional(rollbackFor = RuntimeException.class)
    public void sendMessage2(int num){

        kafkaTemplate.send(TOPIC_NAME,"这个是事务消息 1 i="+num);

        if(num == 0){
            throw new RuntimeException();
        }
        kafkaTemplate.send(TOPIC_NAME,"这个是事务消息 2 i="+num);

    }


    /**
     * 声明式事务
     * @param num
     */
    @GetMapping("/api/v1/tran2")
    public void sendMessage3( int num){

        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, Object, Object>() {
            @Override
            public Object doInOperations(KafkaOperations<String, Object> kafkaOperations) {

                kafkaOperations.send(TOPIC_NAME,"这个是事务消息 1 i="+num);

                if(num == 0){
                    throw new RuntimeException();
                }
                kafkaOperations.send(TOPIC_NAME,"这个是事务消息 2 i="+num);
                return true;
            }
        });
    }
}
